package com.java110.databus.controller;

import com.alibaba.fastjson.JSONObject;
import com.java110.core.factory.AuthenticationFactory;
import com.java110.dto.reportData.ReportDataDto;
import com.java110.dto.reportData.ReportDataHeaderDto;
import com.java110.utils.kafka.KafkaFactory;
import com.java110.utils.util.Assert;
import com.java110.utils.util.BeanConvertUtil;
import com.java110.utils.util.PayUtil;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;


@RequestMapping(value = "/kafka")
@Controller
public class ProducerController {


    @RequestMapping(value = "/producer", method = RequestMethod.POST)
    public void consume(@RequestBody JSONObject reqJson) {
        Assert.hasKeyAndValue(reqJson, "topic", "请求报文中未包含topic");
        Assert.hasKeyAndValue(reqJson, "communitySecure", "请求报文中未包含communitySecure");
        Assert.hasKeyAndValue(reqJson, "data", "请求报文中未包含data");


        JSONObject jsonObject = reqJson.getJSONObject("data");
        JSONObject  header=jsonObject.getJSONObject("header");
        JSONObject  body=jsonObject.getJSONObject("body");
        generatorProducerSign(header,body,reqJson.getString("communitySecure"));


        try {
            KafkaFactory.sendKafkaMessage( reqJson.getString("topic"),  jsonObject.toJSONString() );
        } catch (Exception e) {
            e.printStackTrace();
        }

    }


    public static void generatorProducerSign(JSONObject header, JSONObject body, String code) {

        String newSign = AuthenticationFactory.md5(header.getString("tranId") + header.getString("reqTime") + body.toJSONString() + code).toLowerCase();
        header.put("sign",newSign);
    }


}
